<?php

class Worker
{
    // worker监听端口的Socket
    protected $mainSocket = null;
    // worker的所有连接
    protected $connections = array();
    // worker的所有读buffer
    protected $recvBuffers = array();
    // 当前处理的fd
    protected $currentDealFd = 0;
    // 从socket读数据超时时间,单位毫秒
    protected $recvTimeOut = 1000;
    // 发送数据超时时间
    protected $sendTimeOut = 1000;
    // 逻辑处理超时时间
    protected $processTimeout = 300000;
    // 是否是长连接
    protected $isPersistentConnection = false;
    // work服务名称
    protected $serverName = '';
    // 状态统计
    protected $statusInfo = array();

    public function __construct($socket = null, $name = '', $recv_timeout = 1000, $process_timeout = 30000, $send_timeout = 1000, $persistent_connection = false)
    {
        // 初始化
        $this->mainSocket = $socket;
        $this->serverName = $name;
        $this->recvTimeOut = $recv_timeout >= 0 ? $recv_timeout : $this->recvTimeOut;
        $this->processTimeout = $process_timeout > 0 ? $process_timeout : $this->processTimeout;
        $this->sendTimeOut = $send_timeout > 0 ? $send_timeout : $this->sendTimeOut;
        $this->isPersistentConnection = (bool)$persistent_connection;

        if ($socket) {
            // 设置监听socket非阻塞
            stream_set_blocking($this->mainSocket, 0);
        }
        // 统计信息
        $this->statusInfo[$name] = array(
            'start_time' => 0,
            'total_request' => 0,
            'client_close' => 0,
            'recv_timeout' => 0,
            'client' => array()
        );
    }

    /**
     * 开始工作
     */
    public function work()
    {
        $this->init();
        // 初始化事件轮询库
        $this->event = new Select();

        if ($this->mainSocket) {
            // 添加accept事件
            $this->event->add($this->mainSocket, BaseEvent::EV_ACCEPT, array($this, 'accept'));
        }
        $this->event->loop();
    }

    public function init()
    {
        $bootstrap = "/Users/jiedian-long/Documents/server_test/rpc/{$this->serverName}/index.php";
        if (is_file($bootstrap)) {
            require_once $bootstrap;
        }
    }

    /**
     * 接收一个连接
     *
     * @param $socket
     * @return bool
     */
    public function accept($socket)
    {
        $new_connection = @stream_socket_accept($socket, 0);
        if (false === $new_connection) {
            return false;
        }

        $this->statusInfo[$this->serverName]['total_request']++;

        $fd = (int)$new_connection;
        $this->connections[$fd] = $new_connection;
        $this->statusInfo[$this->serverName]['client'][] = $fd;
        // 非阻塞
        stream_set_blocking($this->connections[$fd], 0);
        $this->event->add($this->connections[$fd], BaseEvent::EV_READ, array($this, 'dealRead'), $fd, $this->recvTimeOut);
    }

    /**
     * 处理收到的数据
     *
     * @param $connection
     * @param $length
     * @param $buffer
     * @param null $fd
     */
    public function dealRead($connection, $length, $buffer, $fd = null)
    {
        $this->currentDealFd = $fd;

        // 出错了
        if ($length == 0) {
            if (feof($connection)) {
                // 客户端提前断开链接
                $this->statusInfo[$this->serverName]['client_close']++;
            } else {
                // 超时了
                $this->statusInfo[$this->serverName]['recv_timeout']++;
            }
            $this->closeClient($fd);
            return;
        }

        if (isset($this->recvBuffers[$fd])) {
            $buffer = $this->recvBuffers[$fd] . $buffer;
        }

        $remain_len = $this->dealInput($buffer);
        // 包接收完毕
        if (0 === $remain_len) {
            try {
                // 业务处理
                $this->dealProcess($buffer);
            } catch (Exception $e) {
                $this->sendToClient($e->getMessage());
            }

            // 是否是长连接
            if ($this->isPersistentConnection) {
                // 清空缓冲buffer
                unset($this->recvBuffers[$fd]);
            } else {
                // 关闭链接
                $this->closeClient($fd);
            }
//            print_r($this->statusInfo);
        } else if (false === $remain_len) {
            // 出错
            $this->closeClient($fd);
        } else {
            // 还有数据没收完，则保存收到的数据，等待其它数据
            $this->recvBuffers[$fd] = $buffer;
        }
    }

    public function dealInput($buffer)
    {
        return 0;
    }

    /**
     * 关闭连接
     *
     * @param int $fd
     * @return void
     */
    protected function closeClient($fd)
    {
        $this->event->delAll($this->connections[$fd]);
        fclose($this->connections[$fd]);

        unset($this->connections[$fd], $this->recvBuffers[$fd]);
    }

    /**
     * 业务逻辑处理
     * @param string $recv_str 收到的数据包
     * @return void
     */
    public function dealProcess($recv_str)
    {
        $request = json_decode($recv_str, true);
        $data = $request['data'];
        $rpcArr = json_decode($data, true);

        try {
            $class = '\\Controller\\' . $rpcArr['class'] . 'Controller';
            if (!class_exists($class)) {
                $sendData['code'] = '404';
                $sendData['msg'] = 'controller not found';
                $this->sendToClient(json_encode($sendData));
            }
            $method = $rpcArr['method'];
            $params = $rpcArr['params'];

            $sendData = [
                'code' => 0,
                'msg' => '',
                'data' => null
            ];
            if (!method_exists($class, $method)) {
                $sendData['code'] = '404';
                $sendData['msg'] = 'method not found';
                $this->sendToClient(json_encode($sendData));
            }

            $ret = call_user_func_array(array(new $class, $method), $params);
            $sendData['data'] = $ret;
            $this->sendToClient(json_encode($sendData));
        } catch (\Exception $e) {
            $sendData['code'] = '500';
            $sendData['msg'] = $e->getMessage();
            $this->sendToClient(json_encode($sendData));
        }
    }

    /**
     * 发送数据到客户端
     *
     * @param $str_to_send
     * @return bool
     */
    public function sendToClient($str_to_send)
    {
        if (!isset($this->connections[$this->currentDealFd])) {
            return false;
        }
        $send_len = @fwrite($this->connections[$this->currentDealFd], $str_to_send);
        if ($send_len == strlen($str_to_send)) {
            return true;
        }
        // 客户端关闭
        if (feof($this->connections[$this->currentDealFd])) {
            echo "client close";
            return false;
        }
        return false;
    }
}